package com.bigdata.hbasedemo.consumer;

import com.bigdata.hbasedemo.producer.Commons;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Properties;

public class Flink2HbaseConsumer {
    private static final Logger logger = LogManager.getLogger(Flink2HbaseConsumer.class);

    public static void main(final String... args) {
        // 创建flink流处理环境
        final StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Commons.EG_KAFKA_SERVER);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "FlinkConsumerGroup");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        senv.enableCheckpointing(1000);
        senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        senv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        DataStream<String> stream = senv.addSource(
                new FlinkKafkaConsumer<>(
                        Commons.EG_KAFKA_TOPIC, new SimpleStringSchema(), props));
                //.setParallelism(1);
        //stream.print();
        stream.addSink(new HbaseWriter());

        try{
            senv.execute("Flink2HbaseConsumer");
        }catch (Exception e){
            logger.error("An error occurred.", e);
        }

    }
}
